/**
 * 
 */
package com.myworld.config.mq.rocket.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** 
 * @ClassName: RockerJFConsumer 
 * @Description: 封装消息消费者
 * @author zhuqiang
 * @date 2018年12月29日 下午1:41:58 
 */
public class RockerJFConsumer {
	/**
	 * 日志对象
	 */
	static final Logger log=LoggerFactory.getLogger(RockerJFConsumer.class);
	/**
	 * RocketMQ消费者（push）
	 */
	private DefaultMQPushConsumer consumer;
	/**
	 * 消费者集群的名称
	 */
	private String consumerGroup;
	/**
	 * 消费者所连接的NameService地址
	 */
	private String namesrvAddr;
	/**
	 * 消费的消息标题
	 */
	private String tag;
	/**
	 * 消费的消息主题
	 */
	private String topic;
	/**
	 * 消费者开启的最小线程数
	 */
	private int consumeThreadMin;
	/**
	 * 消费者开启的最大线程数
	 */
	private int    consumeThreadMax;
	/**
	 * 消息监听器
	 */
	private MessageListenerConcurrently messageListener;
	/**
	 * 构造方法
	 * @param consumerGroup   消费者集群的名称
	 * @param namesrvAddr     消费者所连接的NameService地址
	 * @param tag             消费的消息标题
	 * @param topic           消费的消息主题
	 * @param messageListener 消息监听器
	 */
	public RockerJFConsumer(String consumerGroup,String namesrvAddr, String tag,String 
		topic , MessageListenerConcurrently messageListener,int consumeThreadMin,int consumeThreadMax){
		this.consumerGroup=consumerGroup;
		this.namesrvAddr=namesrvAddr;
		this.tag=tag;
		this.topic=topic;
		this.messageListener=messageListener;
		this.consumeThreadMin=consumeThreadMin;
		this.consumeThreadMax=consumeThreadMax;
	}
	public void init(){
		log.info("start init "+consumerGroup);
		consumer = new DefaultMQPushConsumer(consumerGroup);                        //设置消费者集群的名称
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);   //从队列头部开始消费
		log.info(namesrvAddr);
		consumer.setNamesrvAddr(namesrvAddr);                                       //设置消费者所连接的NameService地址
		consumer.setVipChannelEnabled(false);
		consumer.setConsumeThreadMin(consumeThreadMin);
		consumer.setConsumeThreadMax(consumeThreadMax);
		try {
			consumer.subscribe(topic, tag);                                         //设置消费的消息主题及标题
			consumer.registerMessageListener(messageListener);                      //设置消息监听器
			log.info(consumerGroup+" init success.");
		} catch (MQClientException e) {
			log.info(consumerGroup+" init faild.");
		}
		start();
	}
	public void start(){
		log.info("start  "+consumerGroup+"...");
		try {
			consumer.start();                                                        //开启消费者
			log.info(consumerGroup+" start success.");
		} catch (MQClientException e) {
			e.printStackTrace();
			log.info(consumerGroup+" start faild.");
		}
	}
	public void destroy(){
		consumer.shutdown();                                                         //关闭消费者
	}

}
